package com.example.demo.service.impl;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.example.demo.entity.User;
import com.example.demo.mapper.UserMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * @Author CHAS
 * 2024/8/7 23:55
 **/
@Service
@Slf4j
public class KafkaServiceImpl {

    private static int count = 0;
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;
    @Resource
    private UserMapper userMapper;
    @Resource
    private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
    public void sendMessage(String topic, User user) {
        kafkaTemplate.send(topic, JSONUtil.toJsonStr(user));
        log.info("send message to topic:{}, message:{}", topic, user);
    }
    @KafkaListener(topics = "your-topic-name", groupId = "TMS铭舜-group",id = "TMS铭舜-listener")
    public void listen(String message) {
        count++;
        System.out.println(count);
        System.out.println("Received Message: " + count + " " + message.toString());
        // 这里可以根据接收到的消息做相应的处理，比如调用mapper方法更新数据库等
        User user = JSONUtil.toBean(message, User.class);
        userMapper.insert(user);
        // 在这里处理接收到的消息逻辑
        //kafka压力测试100000级出现了错误，只录入了7721条数据，其他都出错了

    }

    public void stopConsume() {
        kafkaListenerEndpointRegistry.getListenerContainer("TMS铭舜-listener").stop();
    }

    public void startConsume() {
        kafkaListenerEndpointRegistry.getListenerContainer("TMS铭舜-listener").start();
    }
}
